package function;

import common.Constant;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;
import util.DimUtil;
import util.JDBCUtil;
import util.RedisUtil;
import util.ThreadPoolUtil;

import java.sql.Connection;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DIMAsyncFunction<T> extends RichAsyncFunction<T, T> {

    private ThreadPoolExecutor threadPool;
    private Connection conn;

    //异步调用,每碰到一条数据,flink会自动调用这个方法,把业务逻辑放到这里
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        //Phoenix客户端目前不支持异步方式进行读,所以我们需要自己使用线程池来完成这个异步的请求
        //把input的维度补齐之后,就可以把他交给resultFuture,这个数据就会进入流的下个阶段
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                // dim查询工作
                // 因为redis的jedis线程池客户端有bug,所以应该一个线程的请求使用一个redis客户端, 否则会出现问题
                Jedis jedis = RedisUtil.getRedisClient();
                // 具体的维度补充的业务, 这个业务是比较抽象的, 和使用者有关系
                try {
                    addDim(conn, jedis, input, resultFuture);
                } catch (Exception e) {
                    e.printStackTrace();
                }

                jedis.close();
            }
        });
    }

    public abstract void addDim(Connection conn, Jedis jedis, T input, ResultFuture<T> resultFuture) throws Exception;

    @Override
    public void close() throws Exception {
        if (threadPool != null) {
            threadPool.shutdown();
        }
        if (conn != null) {
            conn.close();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
        conn = JDBCUtil.getJDBCConnection(Constant.PHOENIX_URL, Constant.PHOENIX_DRIVER);
    }


}
